package org.example.api.tableapi;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.example.api.bean.SensorReading;

/**
 * @author huangqihan
 * @date 2021/3/5
 */
public class TableApiTest {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 基于数据流创建表
        Table table = tableEnv.fromDataStream(dataStream);

        // 调用 table api
        Table resultTable = table.select("id,temperature")
                .where("id = 'sensor_1'");

        // 执行 sql
        tableEnv.createTemporaryView("sensor", table);
        String sql = "select id,temperature from sensor where id = 'sensor_1'";
        Table resultTable1 = tableEnv.sqlQuery(sql);

        DataStream<Row> rows = tableEnv.toAppendStream(resultTable, Row.class);
        rows.print("rows");

        DataStream<Row> rows1 = tableEnv.toAppendStream(resultTable1, Row.class);
        rows1.print("rows1");

        env.execute();

    }
}
